package cn.net.dpi.wlqc.server.action;

import cn.hutool.extra.spring.SpringUtil;
import cn.net.dpi.wlqc.server.handler.Request;
import cn.net.dpi.wlqc.server.handler.Response;
import cn.net.dpi.wlqc.server.model.dto.PointDTO;
import cn.net.dpi.wlqc.server.model.dto.TrackMessageDTO;
import cn.net.dpi.wlqc.server.proto.ThirdTrackBatch;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.handler.codec.http.FullHttpRequest;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.List;
import java.util.stream.Collectors;

/**
 * 翼卡轨迹上传处理
 *
 * @author Looly
 */
@Slf4j
public class ECarTrackUploadAction implements Action {

    private final KafkaTemplate<String, String> kafkaTemplate = SpringUtil.getBean(KafkaTemplate.class);
    private final ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);

    @Override
    public void doAction(Request req, Response response) {
        ThirdTrackBatch.ThirdTrackResp.Builder respProto = ThirdTrackBatch.ThirdTrackResp.newBuilder()
                .setMsg("success")
                .setCode(0);
        try {
            FullHttpRequest r = (FullHttpRequest) req.getNettyRequest();
            ByteBuf content = r.content();
            // 解析Protobuf
            ThirdTrackBatch.ThirdTrackReq request = ThirdTrackBatch.ThirdTrackReq.parseFrom(
                    new ByteBufInputStream(content));
            List<ThirdTrackBatch.ThirdTrackInfo> tracksList = request.getTracksList();
            if (tracksList.isEmpty()) {
                return;
            }
            tracksList.forEach(track -> {
                try {
                    String deviceId = track.getDeviceId();
                    TrackMessageDTO trackMessageDTO = new TrackMessageDTO();
                    trackMessageDTO.setVer(100);
                    trackMessageDTO.setSource(1);
                    trackMessageDTO.setDeviceId(deviceId);
                    List<ThirdTrackBatch.ThirdPoints> pointsList = track.getPointsList();
                    List<PointDTO> pointDTOList = pointsList.stream().map(this::getPoint).collect(Collectors.toList());
                    trackMessageDTO.setPoints(pointDTOList);
                    String json = objectMapper.writeValueAsString(trackMessageDTO);
                    kafkaTemplate.send("track", trackMessageDTO.getDeviceId(), json);
                } catch (JsonProcessingException e) {
                    log.error("轨迹序列化或发送失败", e);
                }
            });
        } catch (Exception e) {
            respProto.setCode(1);
            log.error(e.getMessage(), e);
            respProto.setMsg(String.format("fail:%s", e.getMessage()));
        }
        byte[] respBytes = respProto.build().toByteArray();
        // 设置响应头
        response.setContentType("application/x-protobuf");
        response.setContentLength(respBytes.length);
        response.setContent(respBytes);
    }

    private PointDTO getPoint(ThirdTrackBatch.ThirdPoints thirdPoints) {
        PointDTO pointDTO = new PointDTO();
        pointDTO.setSTm(thirdPoints.getSystemTime());
        pointDTO.setGTm(thirdPoints.getGpsTime());
        pointDTO.setLon(thirdPoints.getLongitude());
        pointDTO.setLat(thirdPoints.getLatitude());
        pointDTO.setAlt(thirdPoints.getAltitude());
        pointDTO.setV(thirdPoints.getSpeed());
        pointDTO.setDir(thirdPoints.getGpsBearing());
        pointDTO.setCType(thirdPoints.getCoordinateType().getNumber());
        return pointDTO;
    }

}